package person.tavion.springbootkafkademo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

@RestController

public class KafkaTestController {
    @Autowired
    private KafkaProducer1 kafkaProducer;
    @Autowired
    private KafkaConsumer1 kafkaConsumer;
    @Autowired
    private KafkaConnectionDetails kafkaConnectionDetails;
    @RequestMapping("/test")
    public String test(String topic,String msg){
        try {
            kafkaProducer.sendMessage(topic,msg);
        } catch (Exception e) {
            e.printStackTrace();
            return "failed";
        }

        return "success";
    }
    // 发送消息
    @RequestMapping("/send")
    public String send(){
        try {
            String topic = "myTopic";
            String msg = "test-";
//            kafkaProducer.sendMessage(topic,1,"sendFail"+ UUID.randomUUID().toString().substring(0,5),msg+1);
//            kafkaProducer.sendMessage(topic,1,"send"+ UUID.randomUUID().toString().substring(0,5),msg+1);
            for (int i = 0; i < 100000; i++) {
                kafkaProducer.sendMessage(topic,0,"send"+0,msg+i);
                kafkaProducer.sendMessage(topic,1,"send"+1,msg+i);
                kafkaProducer.sendMessage(topic,2,"send"+2,msg+i);
                if(i%1000==0){
                    System.out.println("发送"+i);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            return "failed";
        }

        return "success";
    }
    // 发送会导致死信的消息
    @RequestMapping("/sendFail")
    public String sendFail(){
        try {
            String topic = "myTopic";
            String msg = "test-";
            kafkaProducer.sendMessage(topic,1,"sendFail"+ UUID.randomUUID().toString().substring(0,5),msg+1);
            kafkaProducer.sendMessage(topic,1,"send"+ UUID.randomUUID().toString().substring(0,5),msg+1);
//            for (int i = 0; i < 100000; i++) {
//                kafkaProducer.sendMessage(topic,0,"send"+0,msg+i);
//                kafkaProducer.sendMessage(topic,1,"send"+1,msg+i);
//                kafkaProducer.sendMessage(topic,2,"send"+2,msg+i);
//                if(i%1000==0){
//                    System.out.println("发送"+i);
//                }
//            }


        } catch (Exception e) {
            e.printStackTrace();
            return "failed";
        }

        return "success";
    }
    @RequestMapping("/getDetails")
    public String getDetails(){
        System.out.println(kafkaConnectionDetails.toString());

        return kafkaConnectionDetails.getBootstrapServers().toString();
    }
}
